package com.atguigu.tingshu.search.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch._types.query_dsl.Operator;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.CompletionSuggestOption;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.core.search.HitsMetadata;
import co.elastic.clients.elasticsearch.core.search.Suggestion;
import com.atguigu.tingshu.album.client.AlbumInfoFeignClient;
import com.atguigu.tingshu.album.client.CategoryFeignClient;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.result.Result;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.BaseCategory3;
import com.atguigu.tingshu.model.album.BaseCategoryView;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.model.search.SuggestIndex;
import com.atguigu.tingshu.vo.base.user.UserInfo;
import com.atguigu.tingshu.query.search.AlbumIndexQuery;
import com.atguigu.tingshu.search.service.SearchService;
import com.atguigu.tingshu.user.client.UserInfoFeignClient;
import com.atguigu.tingshu.vo.album.AlbumDetailVo;
import com.atguigu.tingshu.vo.album.AlbumListVo;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.atguigu.tingshu.vo.search.AlbumInfoIndexVo;
import com.atguigu.tingshu.vo.search.AlbumSearchResponseVo;
import com.google.common.collect.Lists;
import jakarta.annotation.Resource;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.data.elasticsearch.client.elc.ElasticsearchTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;

@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class SearchServiceImpl implements SearchService {
    @Resource
    private CategoryFeignClient categoryFeignClient;
    @Resource
    private ElasticsearchClient elasticsearchClient;
    @Resource
    private ElasticsearchTemplate elasticsearchTemplate;
    @Resource
    private AlbumInfoFeignClient albumInfoFeignClient;
    @Resource
    private UserInfoFeignClient userInfoFeignClient;
    @Resource
    private ExecutorService executorService;
    @Resource
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public List<Map<String, Object>> channel(Long category1Id) {
        Result<List<BaseCategory3>> topBaseCategory3R = categoryFeignClient.findTopBaseCategory3(category1Id);
        Assert.notNull(topBaseCategory3R, "首页数据初始化时，根据1级分类id查询三级分类失败！");
        List<BaseCategory3> baseCategory3List = topBaseCategory3R.getData();
        if (CollectionUtils.isEmpty(baseCategory3List)) return null;
        //从集合中取出3级id，封装成FieldValue 以便后续封装DSL语句
        List<FieldValue> category3Ids = baseCategory3List.stream().map(BaseCategory3 ->
                FieldValue.of(BaseCategory3.getId())
        ).toList();
        //将list集合转换成以id为键，单个对象为值的map集合
        Map<Long, BaseCategory3> baseCategory3Map = baseCategory3List.stream()
                .collect(Collectors.toMap(BaseCategory3::getId, baseCategory3 -> baseCategory3));

        try {
            //封装DSL语句
            SearchRequest request = SearchRequest.of(s -> s.index("albuminfo")
                    .query(q -> q.terms(t -> t.field("category3Id").terms(ts -> ts.value(category3Ids))))
                    .aggregations("category3IdAgg", a -> a.terms(t -> t.field("category3Id"))
                            .aggregations("topHotAgg", ag -> ag.topHits(t -> t.size(6)
                                    .sort(st -> st.field(f -> f.field("hotScore").order(SortOrder.Desc)))))));
            //执行DSL语句
            SearchResponse<AlbumInfoIndex> response = elasticsearchClient.search(request, AlbumInfoIndex.class);
            //解析结果
            Map<String, Aggregate> parentAggregations = response.aggregations();
            if (CollectionUtils.isEmpty(parentAggregations)) return null;
            //获取父聚合结果
            Aggregate category3IdAgg = parentAggregations.get("category3IdAgg");
            if (category3IdAgg == null) return null;
            //把桶集合转换成list集合
            List<Map<String, Object>> maps = category3IdAgg.lterms().buckets().array().stream().map(bucket -> {
                Map<String, Object> map = new HashMap<>();
                map.put("baseCategory3", baseCategory3Map.get(bucket.key()));
                //获取子聚合结果
                Map<String, Aggregate> sonAggregations = bucket.aggregations();
                if (CollectionUtils.isEmpty(sonAggregations)) return map;

                Aggregate topHotAgg = sonAggregations.get("topHotAgg");
                if (topHotAgg == null) return map;
                List<AlbumInfoIndex> albumInfoIndices = topHotAgg.topHits().hits().hits().stream()
                        .map(hit -> hit.source().to(AlbumInfoIndex.class)).toList();
                map.put("list", albumInfoIndices);
                return map;
            }).toList();
            return maps;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public AlbumSearchResponseVo search(AlbumIndexQuery query) {
        try {
            //封装DSL语句，获取SearchRequest对象
            SearchRequest request = getSearchRequest(query);
            System.out.println(request);
            //执行DSL语句
            SearchResponse<AlbumInfoIndex> response = elasticsearchClient.search(request, AlbumInfoIndex.class);
            //解析结果，并封装成AlbumSearchResponseVo对象
            AlbumSearchResponseVo searchResponseVo = getAlbumSearchResponseVo(response);
            searchResponseVo.setPageNo(query.getPageNo());
            searchResponseVo.setPageSize(query.getPageSize());

            Long total = searchResponseVo.getTotal();
            Integer pageSize = searchResponseVo.getPageSize();
            searchResponseVo.setTotalPages(total % pageSize == 0 ? total / pageSize : total / pageSize + 1);
            return searchResponseVo;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public List<AlbumInfoIndexVo> findRankingList(Long category1Id, String field) {
        try {
            //封装DSL语句
            SearchRequest request = SearchRequest.of(s -> s.index("albuminfo")
                    .query(q -> q.bool(b -> b.must(m -> m.term(t -> t.field("category1Id").value(category1Id)))))
                    .sort(so -> so.field(f -> f.field(field).order(SortOrder.Desc)))
                    .size(10));
            //执行DSL语句
            SearchResponse<AlbumInfoIndex> response = elasticsearchClient.search(request, AlbumInfoIndex.class);
            //解析结果
            List<Hit<AlbumInfoIndex>> hitList = response.hits().hits();
            if (!CollectionUtils.isEmpty(hitList)) {
                List<AlbumInfoIndexVo> albumInfoIndexVoList = hitList.stream().map(hit -> {
                    AlbumInfoIndex albumInfoIndex = hit.source();
                    AlbumInfoIndexVo albumInfoIndexVo = new AlbumInfoIndexVo();
                    if (albumInfoIndex != null) {
                        BeanUtils.copyProperties(albumInfoIndex, albumInfoIndexVo);
                    }
                    return albumInfoIndexVo;
                }).toList();
                return albumInfoIndexVoList;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public AlbumDetailVo searchAlbumDetail(Long albumInfoId) {
        AlbumDetailVo albumDetailVo = new AlbumDetailVo();

        CompletableFuture<AlbumInfo> async1 = CompletableFuture.supplyAsync(() -> {
            Result<AlbumInfo> albumInfoR = albumInfoFeignClient.getAlbumInfo(albumInfoId);
            Assert.notNull(albumInfoR, "专辑详情展示时，查询专辑基本信息失败！");
            AlbumInfo albumInfo = albumInfoR.getData();
            if (albumInfo == null) return null;
            albumDetailVo.setAlbumInfo(albumInfo);
            return albumInfo;
        }, executorService);

        CompletableFuture<Void> async1_1 = async1.thenAcceptAsync(albumInfo -> {
            Result<UserInfo> userInfoR = userInfoFeignClient.getUserInfo(albumInfo.getUserId());
            Assert.notNull(userInfoR, "专辑详情展示时，查询用户信息失败！");
            UserInfo userInfo = userInfoR.getData();
            if (userInfo == null) return;
            albumDetailVo.setAnnouncer(userInfo);
        }, executorService);

        CompletableFuture<Void> async1_2 = async1.thenAcceptAsync(albumInfo -> {
            Result<BaseCategoryView> baseCategoryViewR = categoryFeignClient.findBaseCategoryView(albumInfo.getCategory3Id());
            Assert.notNull(baseCategoryViewR, "专辑详情展示时，查询分类信息失败！");
            BaseCategoryView baseCategoryView = baseCategoryViewR.getData();
            if (baseCategoryView == null) return;
            albumDetailVo.setBaseCategoryView(baseCategoryView);
        }, executorService);

        CompletableFuture<Void> async2 = CompletableFuture.runAsync(() -> {
            Result<AlbumListVo> albumListVoR = albumInfoFeignClient.findUserAlbum(albumInfoId);
            Assert.notNull(albumListVoR, "专辑详情展示时，查询专辑列表失败！");
            AlbumListVo albumListVo = albumListVoR.getData();
            if (albumListVo == null) return;
            AlbumStatVo albumStatVo = new AlbumStatVo();
            BeanUtils.copyProperties(albumListVo, albumStatVo);
            albumDetailVo.setAlbumStatVo(albumStatVo);
        }, executorService);

        CompletableFuture.allOf(async1, async1_1, async1_2, async2).join();

        return albumDetailVo;
    }

    @Override
    public List<String> completeSuggest(String keyword) {
        try {
            if (keyword == null) return null;
            //封装DSL语句
            SearchRequest request = SearchRequest.of(s -> s.index("suggestinfo")
                    .suggest(su -> su
                            .suggesters("keywordSuggest", sg -> sg.prefix(keyword).completion(c -> c.field("keyword")
                                    .size(10).skipDuplicates(true)))
                            .suggesters("keywordPinyinSuggest", sg -> sg.prefix(keyword).completion(c -> c.field("keywordPinyin")
                                    .size(10).skipDuplicates(true)))
                            .suggesters("keywordSequenceSuggest", sg -> sg.prefix(keyword).completion(c -> c.field("keywordSequence")
                                    .size(10).skipDuplicates(true)))));
            //执行DSL语句
            SearchResponse<SuggestIndex> response = elasticsearchClient.search(request, SuggestIndex.class);
            //解析结果集
            Map<String, List<Suggestion<SuggestIndex>>> suggestMap = response.suggest();
            //获取关键字提示词
            List<String> keywordSuggestList = getStrings(suggestMap, "keywordSuggest");
            //获取关键字拼音提示词
            List<String> keywordPinyinSuggestList = getStrings(suggestMap, "keywordPinyinSuggest");
            //获取关键字首字母提示词
            List<String> keywordSequenceSuggestList = getStrings(suggestMap, "keywordSequenceSuggest");
            HashSet<String> set = new HashSet<>();
            set.addAll(keywordSuggestList);
            set.addAll(keywordPinyinSuggestList);
            set.addAll(keywordSequenceSuggestList);
            if (set.size() >= 10) {
                return set.stream().toList();
            }
            //当数据量不足时，进行匹配查询
            SearchResponse<SuggestIndex> searchResponse = elasticsearchClient.search(s -> s.index("suggestinfo")
                            .size(10)
                            .query(q -> q.match(m -> m.field("title").query(keyword))),
                    SuggestIndex.class);
            //解析结果集
            List<Hit<SuggestIndex>> hits = searchResponse.hits().hits();
            if (!CollectionUtils.isEmpty(hits)) {
                hits.forEach(hit -> {
                    SuggestIndex suggestIndex = hit.source();
                    if (suggestIndex != null) {
                        set.add(suggestIndex.getTitle());
                    }
                });
            }
            return set.stream().toList();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public void updateRankingData() {
        String startTime = this.stringRedisTemplate.opsForValue().get(RedisConstant.RANKING_KEY_PREFIX);
        if (StringUtils.isBlank(startTime)) startTime = "2023-12-21 00:00:00";
        String endTime = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
        this.stringRedisTemplate.opsForValue().set(RedisConstant.RANKING_KEY_PREFIX, endTime);
        Result<List<Long>> albumIdsResult = this.albumInfoFeignClient.getChangeAlbum(startTime, endTime);
        Assert.notNull(albumIdsResult, "更新热门数据时，获取统计信息发生变化的专辑id集合失败！");
        List<Long> albumIds = albumIdsResult.getData();
        if (CollectionUtils.isEmpty(albumIds)) return;
        //考虑数据量问题，将集合拆分成小集合
        List<List<Long>> albumIdsList = Lists.partition(albumIds, 100);

        CountDownLatch countDownLatch = new CountDownLatch(albumIdsList.size());
        albumIdsList.forEach(albumIdList -> {
            executorService.execute(() -> {
                Result<List<AlbumStatVo>> changeAlbumStatsResult = this.albumInfoFeignClient.getChangeAlbumStats(albumIdList);
                Assert.notNull(changeAlbumStatsResult, "更新热门数据时，获取变化的统计信息的集合失败！");
                List<AlbumStatVo> albumStatVoList = changeAlbumStatsResult.getData();

                albumStatVoList.forEach(albumStatVo -> {
                    AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();
                    BeanUtils.copyProperties(albumStatVo, albumInfoIndex);
                    albumInfoIndex.setId(albumStatVo.getAlbumId());
                    albumInfoIndex.setHotScore(albumInfoIndex.getPlayStatNum() * 0.1 + albumInfoIndex.getCommentStatNum() * 0.2 + albumInfoIndex.getCommentStatNum() * 0.3 + albumInfoIndex.getBuyStatNum() * 0.4);
                    this.elasticsearchTemplate.update(albumInfoIndex);
                });
                countDownLatch.countDown();
            });
        });
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static List<String> getStrings(Map<String, List<Suggestion<SuggestIndex>>> suggestMap, String keyword) {
        List<CompletionSuggestOption<SuggestIndex>> keywordSuggest = suggestMap.get(keyword).get(0).completion().options();
        List<String> suggestList = new ArrayList<>();
        if (!CollectionUtils.isEmpty(keywordSuggest)) {
            keywordSuggest.forEach(option -> {
                SuggestIndex suggestIndex = option.source();
                if (suggestIndex != null) {
                    suggestList.add(suggestIndex.getTitle());
                }
            });
        }
        return suggestList;
    }

    private static AlbumSearchResponseVo getAlbumSearchResponseVo(SearchResponse<AlbumInfoIndex> response) {
        HitsMetadata<AlbumInfoIndex> hits = response.hits();
        AlbumSearchResponseVo searchResponseVo = new AlbumSearchResponseVo();
        if (hits.total() != null) {
            //设置总行数
            searchResponseVo.setTotal(hits.total().value());
        }

        List<Hit<AlbumInfoIndex>> hitList = hits.hits();
        if (!CollectionUtils.isEmpty(hitList)) {
            //将List<Hit<AlbumInfoIndex>>类型转换为List<AlbumInfoIndexVo>类型
            List<AlbumInfoIndexVo> albumInfoIndexVos = hitList.stream().map(hit -> {
                AlbumInfoIndex albumInfoIndex = hit.source();
                //将AlbumInfoIndex类型转换为AlbumInfoIndexVo类型
                AlbumInfoIndexVo albumInfoIndexVo = new AlbumInfoIndexVo();
                if (albumInfoIndex != null) {
                    BeanUtils.copyProperties(albumInfoIndex, albumInfoIndexVo);
                }
                //设置高亮
                Map<String, List<String>> highlight = hit.highlight();
                if (!CollectionUtils.isEmpty(highlight)) {
                    if (!CollectionUtils.isEmpty(highlight.get("albumTitle"))) {
                        albumInfoIndexVo.setAlbumTitle(highlight.get("albumTitle").get(0));
                    }
                    if (!CollectionUtils.isEmpty(highlight.get("albumIntro"))) {
                        albumInfoIndexVo.setAlbumTitle(highlight.get("albumIntro").get(0));
                    }
                }
                return albumInfoIndexVo;
            }).toList();
            //设置检索出来的商品信息
            searchResponseVo.setList(albumInfoIndexVos);
        }
        return searchResponseVo;
    }

    private static SearchRequest getSearchRequest(AlbumIndexQuery query) {
        SearchRequest.Builder requestBuilder = new SearchRequest.Builder();
        requestBuilder.index("albuminfo");
//封装DSL查询条件
        //1、查询
        //构建bool查询
        BoolQuery.Builder boolQuery = new BoolQuery.Builder();
        //1.1、关键字查询
        String keyword = query.getKeyword();
        if (!StringUtils.isEmpty(keyword)) {
            boolQuery.must(m -> m.multiMatch(mu -> mu.query(keyword)
                    .fields("albumTitle", "albumIntro", "announcerName")
                    .operator(Operator.And)));
            //4、高亮
            requestBuilder.highlight(h -> h.fields("albumTitle", f ->
                    f.preTags("<font style = 'color:red'>").postTags("</font>")));
        }
        //1.2、三级分类id查询
        Long category1Id = query.getCategory1Id();
        if (category1Id != null) {
            boolQuery.filter(f -> f.term(t -> t.field("category1Id").value(category1Id)));
        }
        Long category2Id = query.getCategory2Id();
        if (category2Id != null) {
            boolQuery.filter(f -> f.term(t -> t.field("category2Id").value(category2Id)));
        }
        Long category3Id = query.getCategory3Id();
        if (category3Id != null) {
            boolQuery.filter(f -> f.term(t -> t.field("category3Id").value(category3Id)));
        }
        //1.3、构建嵌套查询
        List<String> attributeList = query.getAttributeList();
        if (!CollectionUtils.isEmpty(attributeList)) {
            attributeList.forEach(attribute -> {
                String[] arr = attribute.split(":");
                if (arr.length == 2) {
                    boolQuery.filter(f -> f.nested(n -> n.path("attributeValueIndexList")
                            .query(q -> q.bool(b -> b.must(m -> m.term(t ->
                                            t.field("attributeValueIndexList.attributeId").value(arr[0])))
                                    .must(m -> m.term(t ->
                                            t.field("attributeValueIndexList.attributeId").value(arr[1])))))));
                }
            });
        }
        //将bool查询放入requestBuilder
        requestBuilder.query(boolQuery.build()._toQuery());
        //2、排序
        String queryOrder = query.getOrder();
        if (!StringUtils.isEmpty(queryOrder)) {
            String[] orders = queryOrder.split(":");
            if (orders.length == 2 && StringUtils.isNumeric(orders[0]) && Arrays.asList("desc", "asc").contains(orders[1])) {
                String orderField = switch (orders[0]) {
                    case "1" -> "hotScore";
                    case "2" -> "playStatNum";
                    case "3" -> "createTime";
                    default -> "";
                };
                requestBuilder.sort(s -> s.field(f -> f.field(orderField).order(orders[1].equals("desc") ? SortOrder.Desc : SortOrder.Asc)));
            }
        }
        //3、分页
        Integer pageNo = query.getPageNo();
        Integer pageSize = query.getPageSize();
        requestBuilder.from((pageNo - 1) * pageSize);
        requestBuilder.size(pageSize);
        //5、结果集过滤(加filter)
        requestBuilder.source(s -> s.filter(f -> f.excludes("createTime", "hotScore")));
        return requestBuilder.build();
    }
}
